package com.cy.improve.demos.web.mq;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class MessageConsumer {

    @Resource
    private ElasticsearchClient esClient;

    public static Integer count = 0;

    @KafkaListener(topics = "businesslog", groupId = "my-group-id")
    public void listen0(List<String> records, Acknowledgment ack) throws IOException {
        processRecords(records, ack);
    }

    @KafkaListener(topics = "businesslog", groupId = "my-group-id")
    public void listen1(List<String> records, Acknowledgment ack) throws IOException {
        processRecords(records, ack);
    }

    @KafkaListener(topics = "businesslog", groupId = "my-group-id")
    public void listen2(List<String> records, Acknowledgment ack) throws IOException {
        processRecords(records, ack);
    }

    @KafkaListener(topics = "businesslog", groupId = "my-group-id")
    public void listen3(List<String> records, Acknowledgment ack) throws IOException {
        processRecords(records, ack);
    }

    @KafkaListener(topics = "businesslog", groupId = "my-group-id")
    public void listen4(List<String> records, Acknowledgment ack) throws IOException {
        processRecords(records, ack);
    }



    private void processRecords(List<String> records, Acknowledgment ack) throws IOException {
        BulkRequest.Builder br = new BulkRequest.Builder();

        for (String record : records) {
            Map<String, Object> document = new ObjectMapper().readValue(record, Map.class);
            br.operations(op -> op
                    .index(idx -> idx
                            .index("business-log")
                            .document(document)
                    )
            );
        }

        BulkResponse result = esClient.bulk(br.build());
        count += records.size();
        log.info("拉取{}条，总存入{}条数据", records.size(), count);

        if (result.errors()) {
            log.error("Bulk had errors");
            for (BulkResponseItem item : result.items()) {
                if (item.error() != null) {
                    log.error(item.error().reason());
                }
            }
        }

        ack.acknowledge();
    }
}